package cn.itcast.Streams;


import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;

import java.util.HashMap;

public class LogProcessor implements Processor<byte[],byte[]> {
    //上下文对象
    private ProcessorContext processorContext;
    @Override
    public void init(ProcessorContext processorContext) {
        //初始化方法
        this.processorContext=processorContext;
    }
    @Override
    public void process(byte[] key, byte[] value) {
        //处理一条消息
        String inputOri = new String(value);
        HashMap <String,Integer> map = new HashMap<String,Integer>();
        int times = 1;
        if(inputOri.contains(" ")){
            //截取字段
            String [] words = inputOri.split(" ");
            for (String word : words){
                if(map.containsKey(word)){
                    map.put(word,map.get(word)+1);
                }else{
                    map.put(word,times);
                }
            }
        }
        inputOri = map.toString();
        processorContext.forward(key,inputOri.getBytes());
    }
    @Override
    public void close() {}
}
